package com.mate.cloud.redisson.event;

import com.google.common.collect.Maps;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * 标签事件处理（使用 Redis Stream）
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/19 10:24
 * @updateUser:     MI
 * @updateTime:     2025/8/19 10:24
 * @updateRemark:   修改内容
 * @version:        1.0
 */
@Service
public class TagEventService {

    private static final String TAG_EVENT_STREAM = "tag:events";

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 发布标签事件
     */
    public void publishTagEvent(String itemId, String tagName, String action) {
        Map<String, Object> eventData = Maps.newHashMap();
        eventData.put("itemId", itemId);
        eventData.put("tag", tagName);
        eventData.put("action", action);
        eventData.put("timestamp", System.currentTimeMillis());

        ObjectRecord<String, Map<String, Object>> record =
                StreamRecords.objectBacked(eventData)
                        .withStreamKey(TAG_EVENT_STREAM);

        redisTemplate.opsForStream().add(record);
    }

    /**
     * 处理标签事件
     */
    @PostConstruct
    public void startEventConsumer() {
        new Thread(() -> {
            while (true) {
                try {
                    List<MapRecord<String, Object, Object>> records =
                            redisTemplate.opsForStream().read(
                                    Consumer.from("tag-service", "instance-1"),
                                    StreamReadOptions.empty().count(10),
                                    StreamOffset.create(TAG_EVENT_STREAM, ReadOffset.lastConsumed())
                            );

                    if (records != null && !records.isEmpty()) {
                        processEvents(records);
                    }
                } catch (Exception e) {
                    // 处理异常
                }
            }
        }).start();
    }

    private void processEvents(List<MapRecord<String, Object, Object>> records) {
        for (MapRecord<String, Object, Object> record : records) {
            Map<Object, Object> event = record.getValue();
            String action = (String) event.get("action");
            String itemId = (String) event.get("itemId");
            String tag = (String) event.get("tag");

            // 根据事件类型处理
            if ("add".equals(action)) {
                // 更新推荐系统
            } else if ("remove".equals(action)) {
                // 清理相关数据
            }

            // 确认消息已处理
            redisTemplate.opsForStream().acknowledge(TAG_EVENT_STREAM, "tag-service", record.getId());
        }
    }
}
